package net.flyinggroup.introduction;

import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.core.RSocketConnector;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.util.DefaultPayload;
import reactor.core.publisher.Flux;

import java.nio.ByteBuffer;

import static net.flyinggroup.introduction.Constants.DATA_STREAM_NAME;
import static net.flyinggroup.introduction.Constants.TCP_PORT;

public class ReqStreamClient {
    private final RSocket socket;

    public ReqStreamClient() {
        socket = RSocketConnector.connectWith(TcpClientTransport.create(TCP_PORT)).block();
    }

    public Flux<Float> getDataStream() {
        return socket.requestStream(DefaultPayload.create(DATA_STREAM_NAME))
                .map(Payload::getData)
                .map(ByteBuffer::getFloat)
                .doOnNext(System.out::println)
                .onErrorReturn(-1f);
    }

    public void dispose() {
        this.socket.dispose();
    }
}
